
package com.springboot.rmq.consumer;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import com.springboot.rmq.comm.ResultCode;
import com.springboot.rmq.consumer.bean.MsgConfigBean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * MsgConsumer:
 * 消息的消费者
 */
public class MsgConsumer {
    /**
     * 日志记录器
     */
    private static final Log log = LogFactory.getLog(MsgConsumer.class);

    /**
     * 记录消息队列请求的次数
     */
    private static volatile AtomicInteger requestCount = new AtomicInteger(0);

    /**
     * 创建DefaultMQPushConsumer对象
     */
    private DefaultMQPushConsumer consumer;

    /**
     * 创建构造函数
     *
     * @param configBean configBean
     */
    public MsgConsumer(MsgConfigBean configBean) {
        //设置nameserverAddr
        initConsumer(configBean);

    }

    /**
     * 设置consumer的基本属性
     *
     * @param configBean confingBean
     */
    private void initConsumer(MsgConfigBean configBean) {
        if (log.isDebugEnabled()) {
            log.debug("ready to create consumer with configuration:[" + configBean + "]");
        }

        this.consumer = new DefaultMQPushConsumer();
        //设置groupname
        this.consumer.setConsumerGroup(configBean.getGroupName());
        //设置nameAddr
        this.consumer.setNamesrvAddr(configBean.getNameAddr());
        //设置instancename
        consumer.setInstanceName(configBean.getInstanceName());
        //this.consumer.setInstanceName(Constant.CONSUMER_INSTANCE_NAME);
        //设置batchMaxSize();
        this.consumer.setConsumeMessageBatchMaxSize(configBean.getConsumeMessageBatchMaxSize());
        //设置最小线程数
        this.consumer.setConsumeThreadMin(configBean.getConsumerThreadCountMin());
        //设置最大线程数
        this.consumer.setConsumeThreadMax(configBean.getConsumerThreadCountMax());

        //固定从消息头开始取数据
        this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        try {
            MsgHandle beanMsgHandle = configBean.getMsgHandle();
            if (beanMsgHandle == null) {
                beanMsgHandle = (MsgHandle) configBean.getHandle().newInstance();
            }

            final MsgHandle msgHandle = beanMsgHandle;
            //无序消息消费模式，则根据默认的函数结果来判断是否成功
            if (!configBean.isOrder()) {
                this.consumer.setMessageListener(new MessageListenerConcurrently() {
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                        logMsg(msgs);


                        MsgConsumer.this.incrementAndGet();

                        String resultCode = msgHandle.dealMessage(msgs);

                        MsgConsumer.this.decrementAndGet();
                        log.info(
                                "Concurrently deal by handle:[" + msgHandle.getClass().getName() + "] with resultcode:["
                                        + resultCode + "]");
                        if (resultCode.equals(ResultCode.CONCURRENT_SUCCESS)) {
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } else {
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                        }
                    }
                });
            } else {
                //顺序消息消费模式，则根据默认的函数结果来判断是否成功
                this.consumer.setMessageListener(new MessageListenerOrderly() {
                    @Override
                    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                        logMsg(msgs);


                        MsgConsumer.this.incrementAndGet();
                        context.setAutoCommit(false);
                        String resultCode = msgHandle.dealMessage(msgs);
                        MsgConsumer.this.decrementAndGet();
                        log.info("message deal by handle:[" + msgHandle.getClass().getName() + "] with resultcode:["
                                + resultCode + "]");
                        switch (resultCode) {
                            case ResultCode.ORDER_SUCCESS:
                                return ConsumeOrderlyStatus.SUCCESS;
                            case ResultCode.ORDER_ROLLBACK:
                                return ConsumeOrderlyStatus.ROLLBACK;

                            case ResultCode.ORDER_COMMIT:
                                return ConsumeOrderlyStatus.COMMIT;
                            default:
                                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                        }
                    }
                });
            }

            } catch (Throwable e) {
                decrementAndGet();
               log.error("generate consumer error." , e);
                return;
            }


        //this.consumer.setMessageListener(configBean.getMsgListener());
        //
        try {
            if("false".equals(configBean.getSelfConsumer())) {
                this.consumer.subscribe(configBean.getToppic(), configBean.getTags());
            }else{//这个消费者只能消费本机消息
                this.consumer.setConsumerGroup(configBean.getGroupName()+System.getProperty("IP"));//topic不一样，必须不同组
                this.consumer.subscribe(configBean.getToppic()+System.getProperty("IP"), configBean.getTags());
            }
            consumer.start();

            log.info(
                "create consumer over and start successful"
                    + " with instancename:[" + consumer.getInstanceName() + "]");
        } catch (Throwable e) {
            log.error(" start or subscribe message info error.", e);
        }

    }

    private void decrementAndGet(){
        requestCount.decrementAndGet();
    }

    private void incrementAndGet(){
        requestCount.incrementAndGet();
    }

    /**
     * 判断消息在关闭之后是否已经完成消费完成，true 完成
     * @return
     */
    public static boolean mqIsComplate(){
        return requestCount.get() <= 0;
    }

    private static void logMsg(List<MessageExt> msgs){
        if (msgs != null && !msgs.isEmpty()) {
            for (MessageExt msg : msgs) {
                log.info("收到rocketmq的消息+++++++:" + msg);
            }
        }
    }

    /**
     * 关闭consumer
     */
    public void shutdown() {
        this.consumer.shutdown();
    }
}
